package main.test

import org.apache.spark.graphx.{Graph, TripletFields}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.graphframes.GraphFrame

object test {
  def main(args: Array[String]): Unit = {
    //  def constructGraph: GraphFrame = {
    //    The following example demonstrates how
    //    to create a GraphFrame from vertex and edge DataFrame
    val sparkConf = new SparkConf()
    sparkConf.setAppName("createGraph").setMaster("local[*]")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new SQLContext(sc)
    val v = sqlContext.createDataFrame(List(
      ("a", "Alice", 34),
      ("b", "Bob", 36),
      ("c", "Charlie", 30),
      ("d", "David", 29),
      ("e", "Esther", 32),
      ("f", "Fanny", 36)
//      ("g", "Gabby", 60)
    )).toDF("id", "name", "age")

    // Edge DataFrame
    val e = sqlContext.createDataFrame(List(
      ("a", "b", "friend"),
      ("b", "c", "follow"),
      ("c", "b", "follow"),
      ("f", "c", "follow"),
      ("e", "f", "follow"),
      ("e", "d", "friend"),
      ("d", "a", "friend")
//      ("a", "e", "friend")
    )).toDF("src", "dst", "relationship")


    // Create a GraphFrame
    val g = GraphFrame(v, e)



    // Run PageRank for a fixed number of iterations
    val pagres2: GraphFrame = g.pageRank.resetProbability(0.15).maxIter(1).run()
    pagres2.edges.show()
    pagres2.vertices.show()

    //Run pagerank personalized for vertex "a"
    val pagres3: GraphFrame = g.pageRank.resetProbability(0.15).maxIter(1).sourceId("a").run()
    pagres3.edges.show()
    pagres3.vertices.show()

    // Initialize the PageRank graph with each edge attribute having
    // weight 1/outDegree and each vertex with attribute 1.0.
    // When running personalized pagerank, only the source vertex
    // has an attribute 1.0. All others are set to 0.
//    var rankGraph: Graph[Double, Double] = g.toGraphX
//      // Associate the degree with each vertex
//      .outerJoinVertices(g.outDegrees) { (vid, vdata, deg) => deg.getOrElse(0) }
//      // Set the weight on the edges based on the degree
//      .mapTriplets( e => 1.0 / e.srcAttr, TripletFields.Src )

  }

}
